package spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object WindowOperations {
  // Todo WindowOperations
  // Window Operations可以设置窗口的大小和滑动窗口的间隔来动态的获取当前Steaming的允许状态。
  // 所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长。
  //    窗口时长：计算内容的时间范围；
  //    滑动步长：隔多久触发一次计算。 (这两者都必须为采集周期大小的整数倍)
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("WindowOperations")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    ssc.checkpoint("./ck")

    // Create a DStream that will connect to hostname:port, like localhost:9999
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    // Split each line into words
    val words: DStream[String] = lines.flatMap(_.split(" "))


    // Count each word in each batch
    val pairs: DStream[(String, Int)] = words.map(word => (word, 1))
    // Todo 关于Window的操作还有如下方法：
    /**   *
     * （1）window(windowLength, slideInterval)
     * : 基于对源 DStream 窗化的批次进行计算返回一个新的Dstream；
     * （2）countByWindow(windowLength, slideInterval)
     * : 返回一个滑动窗口计数流中的元素个数；
     * （3）reduceByWindow(func, windowLength, slideInterval)
     * : 通过使用自定义函数整合滑动区间 流元素来创建一个新的单元素流；
     * （4）reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])
     * : 当在一个(K,V)对的DStream上调用此函数，会返回一个新(K,V)对的DStream，
     * 此处通过对滑动窗口中批次数据使用reduce函数来整合每个key的value值。
     * （5）reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])
     * : 这个函数是上述函数的变化版本，每个窗口的reduce值都是通过用前一个窗的reduce值来 递增 计算。
     * 通过reduce进入到滑动窗口数据并”反向reduce”离开窗口的旧数据来实现这个操作。一个例
     * 子是随着窗口滑动对keys的“加”“减”计数。通过前边介绍可以想到，这个函数只适用于”
     * 可逆的reduce函数”，也就是这些reduce函数有相应的”反reduce”函数(以参数invFunc形式传入)。
     * 如前述函数，reduce任务的数量通过可选参数来配置。
     * countByWindow()和 countByValueAndWindow()作为对数据进行计数操作的简写。
     * countByWindow()返回一个表示每个窗口中元素个数的DStream，
     * 而countByValueAndWindow()返回的DStream则包含窗口中每个值的个数。
     */
    val wordCounts: DStream[(String, Int)] = pairs.reduceByKeyAndWindow(
      // 加上新进入窗口的批次中的元素
      (a, b) => a + b,
      // 移除离开窗口的老批次中的元素
      (a, b) => a - b,
      // 窗口时长
      Seconds(12),
      // 滑动步长
      Seconds(6)
    )

    // Print the first ten elements of each RDD generated in this DStream to the console
    wordCounts.print()

    // Start the computation and wait for termination
    ssc.start()

    ssc.awaitTermination()
  }
}
